package com.nepu.spark.realtime.app

import java.lang
import java.text.SimpleDateFormat
import java.time.{LocalDate, Period}
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import com.nepu.spark.realtime.bean.{DauInfo, PageLog}
import com.nepu.spark.realtime.constant.{DateConstant, KafkaConstant}
import com.nepu.spark.realtime.util._
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.{Jedis, Pipeline}

import scala.collection.mutable.ListBuffer

/**
  * 本类做的是DWD层的日活宽表
  *
  * @author chenshuaijun
  * @create 2022-10-31 15:58
  */
/**
  * 流程梳理：
  * 1、初始化SparkStreaming的上下文环境对象
  * 2、读取offset
  * 3、从kafka中消费数据
  * 4、提取offset
  * 5、转换数据结构
  * 6、处理数据
  *   6.1、去重：自我审查
  *   6.2、去重：第三方审查
  * 7、数据写入到es
  * 8、提交offset
  */
object DwdDailyLivingWideTable {


  def main(args: Array[String]): Unit = {
    // 每次启动任务的时候，需要以es为基准，同步redis中的状态
    consistentState()
    // 1、 初始化上下文环境对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("dwd_daily_living_width_table")
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    // 2、从Redis中读取对应的offset
    // 2.1、因为我们消费的数据是pageLog数据，所以我们要使用ods层日志数据分流的这个topic
    val topic: String = KafkaConstant.DWD_PAGE_LOG
    val groupId: String = "dwd_daily_living_width_table_group"
    val offset: Map[TopicPartition, Long] = KafkaOffsetUtils.getOffset(topic, groupId)
    // 消费数据
    var kafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if (offset != null && offset.nonEmpty) {
      kafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, topic, groupId, offset)
    } else {
      kafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, topic, groupId)
    }

    // 提取offset
    var offsetRanges: Array[OffsetRange] = null
    val offsetRangesDStream: DStream[ConsumerRecord[String, String]] = kafkaDStream.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    // 转换数据结构，这里数据是有具体的bean的
    val mapDStream: DStream[PageLog] = offsetRangesDStream.map(
      consumerRecord => {
        val record: String = consumerRecord.value()
        val pageLog: PageLog = JSON.parseObject(record, classOf[PageLog])
        pageLog
      }
    )
    // 自我进行监视，其实就是过滤掉那些last_page_id不为空的数据，只保留首次登录的数据
    val filterDStream: DStream[PageLog] = mapDStream.filter(pageLog => pageLog.lastPageId == null)
    // 因为统计的指标是日活，所以我们要做的是，一个设备无论在一天中登录了几次我们只记录一次，所以这里使用的使用Redis做第三方的审查
    val mapPartitionDStream: DStream[PageLog] = filterDStream.mapPartitions(
      iteator => {
        // 因为下面要用到Redis，所以我们要在这里获取Redis中的对象
        val jedis: Jedis = MyRedisUtils.getJedisClient()
        // 因为我们的MapperPartition需要返回的是一个迭代器，所以我们将在这里定义一个数组，用来存放通过第三方审查的数据
        val pageLogs: ListBuffer[PageLog] = new ListBuffer[PageLog]()
        // 定义日期转换器
        val sdf = new SimpleDateFormat(DateConstant.YMD)
        for (pageLog <- iteator) {
          // 取出pageLog中的时间戳，因为我们等下模拟的时候，可以产生不同天的数据，所以我们就不能使用当前系统的日期，
          // 但这个只是在我们模拟的项目中使用真实的项目是用不到的。
          val ts: Long = pageLog.ts
          val date = new Date(ts)
          val dateStr: String = sdf.format(date)
          // 定义Redis中存放日活mid的key
          val duaKay: String = s"DAU:$dateStr"
          // 取到对应的mid
          val mid: String = pageLog.mid
          // 判断Redis中是否有这个mid
          val isSuccessFlag: lang.Long = jedis.sadd(duaKay, mid)
          if (isSuccessFlag == 1L) {
            // 表示我们将当前的mid存放到，Redis中成功了。就将当前的pageLog存放到对应的List中
            pageLogs.append(pageLog)
          }
        }
        // 关闭Redis
        jedis.close()
        pageLogs.toIterator
      }
    )
    // 维度合并
    val dimMergeDStream: DStream[DauInfo] = mapPartitionDStream.mapPartitions(
      iter => {
        // 获取jedis的连接，用于后续的join
        val jedis: Jedis = MyRedisUtils.getJedisClient()
        // 日期
        val sdf = new SimpleDateFormat(DateConstant.YMD_HMS)
        // 存放日活数据
        val dauInfos = new ListBuffer[DauInfo]()
        for (pageLog <- iter) {
          // beanCopy
          val dauInfo = new DauInfo()
          MyBeanCopyUtils.fieldCopy(pageLog, dauInfo)
          // 关联用户维度数据
          val userId: String = pageLog.userId
          // 拼装redis中的key
          val userKey: String = s"DIM:USER_INFO:$userId"
          val userInfoJson: String = jedis.get(userKey)
          val userInfoJsonObj: JSONObject = JSON.parseObject(userInfoJson)
          val userGender: String = userInfoJsonObj.getString("gender")
          val birthday: String = userInfoJsonObj.getString("birthday")
          val birthdayDate: LocalDate = LocalDate.parse(birthday)
          val currentDate: LocalDate = LocalDate.now()
          // 因为用户的年龄每年都在变，所以我们的维度表中是不维护对应的年龄的，而是维护出生日期，之后进行计算
          val userAge: String = Period.between(birthdayDate, currentDate).getYears.toString
          dauInfo.user_gender = userGender
          dauInfo.user_age = userAge
          // 关联地区维度数据
          val provinceId: String = pageLog.ar
          // 拼接Redis中的key
          val provinceKey: String = s"DIM:BASE_PROVINCE:$provinceId"
          val provinceJson: String = jedis.get(provinceKey)
          val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)
          val provinceName: String = provinceJsonObj.getString("name")
          val provinceIsoCode: String = provinceJsonObj.getString("iso_code")
          val provinceIso31622: String = provinceJsonObj.getString("iso_3166_2")
          val provinceAreaCode: String = provinceJsonObj.getString("area_code")
          dauInfo.province_name = provinceName
          dauInfo.province_iso_code = provinceIsoCode
          dauInfo.province_3166_2 = provinceIso31622
          dauInfo.province_area_code = provinceAreaCode
          // 处理时间维度数据
          val ts: Long = pageLog.ts
          val date = new Date(ts)
          val dateStr: String = sdf.format(date)
          val dths: Array[String] = dateStr.split(" ")
          dauInfo.dt = dths(0)
          dauInfo.hr = dths(1).split(":")(0)
          // 放入集合中
          dauInfos.append(dauInfo)
        }
        // 关闭jedis的连接
        jedis.close()
        dauInfos.toIterator
      }
    )
    // 将数据存储到es中
    dimMergeDStream.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          iter => {
            val list: List[DauInfo] = iter.toList
            val dauInfos: List[(String, DauInfo)] = list.map(dauInfo => (dauInfo.mid, dauInfo))
            if (dauInfos.nonEmpty) {
              // 取出第一条数据
              val headDau: (String, DauInfo) = dauInfos.head
              // 取出时间戳
              val ts: Long = headDau._2.ts
              val date: String = MyDateUtils.parseTs(ts, DateConstant.YMD)
              // 封装es的key
              val esKey: String = s"gmall_dau_info_$date"
              // 将数据存入到es中
              MyElasticSearchUtils.batchIdempotentWrite(esKey, dauInfos)
            }
          }
        )
        KafkaOffsetUtils.saveOffset(topic, groupId, offsetRanges)
      }
    )

    ssc.start()
    ssc.awaitTermination()
  }

  def consistentState(): Unit = {
    // 拿到当前的日期
    val date: LocalDate = LocalDate.now()
    // 封装索引
    val indexName: String = s"gmall_dau_info_$date"
    val fieldName: String = "mid"
    // 先读取ES中的Mid
    val mids: List[String] = MyElasticSearchUtils.searchField(indexName, fieldName)
    // 将数据写入到Redis
    val jedis: Jedis = MyRedisUtils.getJedisClient()
    // 首先删除Redis中对应的key
    // redis的key
    val redisKey: String = s"DAU:$date"
    jedis.del(redisKey)
    // 判断返回过来的数据
    if (mids != null && mids.nonEmpty) {
      val pipeline: Pipeline = jedis.pipelined()
      for (mid <- mids) {
        pipeline.sadd(redisKey, mid)
      }
      pipeline.sync()
    }
    jedis.close()
  }
}
